package cn.xu.framework.concurrent.queue;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author xuguofei
 * @Date 2024/9/4
 * @Desc TODO
 **/
@Slf4j
public abstract class AbstractDistributeRouteQueueMonitorHandle implements InitializingBean {
    private static ScheduledExecutorService executor = null;

    private volatile static AtomicInteger alarmFalseCount = new AtomicInteger(0);

    @Value("${distributeRouteQueueMonitorHandle.alarmIgnoreRunCount:3}")
    protected Integer alarmIgnoreRunCount;

    @Value("${distributeRouteQueueMonitorHandle.monitorOnOff:true}")
    protected Boolean monitorOnOff;

    @Value("${distributeRouteQueueMonitorHandle.take.handle.maxRetryCount:3}")
    protected Integer maxRetryCount;

    @Value("${distributeRouteRedisLuaQueue.queue.max.size:50000}")
    private Integer queueMaxSize;

    @Value("${distributeRouteRedisLuaQueue.queue.max.size.alarm.rate:0.6}")
    private Double queueMaxSizeAlarmRate;

    private final static String ALARM_IGNORE_KEY = "queue_not_alarm_sleep";

    private static Cache<String, String> alarmIgnoreCache = CacheBuilder.newBuilder().maximumSize(1) // 设置缓存的最大容量
            .expireAfterWrite(60, TimeUnit.SECONDS) //
            .concurrencyLevel(Runtime.getRuntime().availableProcessors()) // 设置并发级别为cpu核心数
            .recordStats() // 开启缓存统计
            .build();

    @Override
    public void afterPropertiesSet() throws Exception {
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(() -> {
            if(alarmFalseCount.get()>=alarmIgnoreRunCount){
                alarmFalseCount.set(0);
                alarmIgnoreCache.put(ALARM_IGNORE_KEY, "1");
            }
            if(StringUtils.isNotBlank(alarmIgnoreCache.getIfPresent(ALARM_IGNORE_KEY))) {
                return;
            }
            boolean alarmFlag = AbstractDistributeRouteQueueMonitorHandle.this.monitor();
            if(alarmFlag){
                alarmFalseCount.addAndGet(1);
            }
        }, 5000, 10000, TimeUnit.MILLISECONDS);
    }

    public boolean monitor() {
        if(!monitorOnOff){
            return true;
        }
        Long length = this.queueMaxCurrentLength();
        if(length!=null&&length>0){
            double rate = length/queueMaxSize;
            if(rate>=queueMaxSizeAlarmRate){
                log.warn("redisQueue,queueMaxLength,{}", length);
                this.afterMonitorHandle(rate, queueMaxSizeAlarmRate, queueMaxSize);
                return false;
            }
        }
        return true;
    }

    protected abstract void afterMonitorHandle(Double rate, Double queueMaxSizeAlarmRate, Integer queueMaxSize);

    protected abstract Long queueMaxCurrentLength();
}
